#include "nwtcpclient.h"
#include <chrono>

#if 0
using asio::ip::tcp;
using tcp_socket = asio::basic_stream_socket<asio::ip::tcp, asio::io_context::executor_type>;
struct close_after
{
    close_after(std::chrono::steady_clock::duration t, tcp_socket& s)
        : timeout_(t), socket_(s)
    {
    }

    // The maximum time to wait for an asynchronous operation to complete.
    std::chrono::steady_clock::duration timeout_;

    // The socket to be closed if the operation does not complete in time.
    tcp_socket& socket_;
};

namespace asio {

// The async_result template is specialised to allow the close_after token to
// be used with asynchronous operations that have a completion signature of
// void(error_code, T). Generalising this for all completion signature forms is
// left as an exercise for the reader.
template <typename T>
class async_result<close_after, void(std::error_code, T)>
{
    public:
    // The initiate() function is used to launch the asynchronous operation by
    // calling the operation's initiation function object. For the close_after
    // completion token, we use this function to run the io_context until the
    // operation is complete.
    template <typename Init, typename... Args>
    static T initiate(Init init, close_after token, Args&&... args)
    {
        asio::io_context& io_context = asio::query(
                    token.socket_.get_executor(), asio::execution::context);

        // Call the operation's initiation function object to start the operation.
        // A lambda is supplied as the completion handler, to be called when the
        // operation completes.
        std::error_code error;
        T result;
        init([&](std::error_code e, T t)
        {
            error = e;
            result = t;
        }, std::forward<Args>(args)...);

        // Restart the io_context, as it may have been left in the "stopped" state
        // by a previous operation.
        io_context.restart();

        // Block until the asynchronous operation has completed, or timed out. If
        // the pending asynchronous operation is a composed operation, the deadline
        // applies to the entire operation, rather than individual operations on
        // the socket.
        io_context.run_for(token.timeout_);

        // If the asynchronous operation completed successfully then the io_context
        // would have been stopped due to running out of work. If it was not
        // stopped, then the io_context::run_for call must have timed out and the
        // operation is still incomplete.
        if (!io_context.stopped())
        {
            // Close the socket to cancel the outstanding asynchronous operation.
            token.socket_.close();

            // Run the io_context again until the operation completes.
            io_context.run();
        }

        // If the operation failed, throw an exception. Otherwise return the result.
        return error ? throw std::system_error(error) : result;
    }
};

} // namespace asio
#endif

namespace NetworkUtil {



TcpClient::TcpClient(const ConnectionConfigInfo& stConnCfgInfo)
    : BaseConnection(stConnCfgInfo)
{

}

TcpClient::~TcpClient()
{
    m_pfnRecvCallback = nullptr;
}

void TcpClient::setRecvCallbackFunc(std::function<void(const std::string& strDataBuf)> pfnRecvCallback)
{
    m_pfnRecvCallback = pfnRecvCallback;
    return;
}

void TcpClient::init()
{
    if (m_bInited)
    {
        return;
    }

    if (!m_pReadBuffer) {
        m_pReadBuffer = new (std::nothrow) char[m_stConnCfgInfo.ui32BufMaxSize];
    }

    if (!connect()) {
        m_bInited = true;
        deinit();

        if (m_stConnCfgInfo.bAutoReconn) {
            reconnectToServer();
        }
    }

    return;
}

void TcpClient::deinit()
{
    if (m_bInited)
    {
        try {
            if (m_sp4ObjSocket.get()) {
                //asio::error_code ignored_ec;
                //m_sp4ObjSocket->shutdown(asio::ip::udp::socket::shutdown_both, ignored_ec);
                //m_sp4ObjSocket->release();
                //m_sp4ObjSocket->close(ignored_ec);

                if (m_bConnected) {
                    m_sp4ObjSocket->shutdown(asio::ip::udp::socket::shutdown_both);
                    m_sp4ObjSocket->release();
                    m_sp4ObjSocket->close();
                }

                m_sp4ObjSocket.reset();
                m_sp4ObjSocket = nullptr;
            }

            if (m_sp4ObjIOCtx.get()) {
                m_sp4ObjIOCtx->stop();
                m_sp4ObjIOCtx.reset();
                m_sp4ObjIOCtx = nullptr;
            }
        } catch (...) {
            // Exception
        }

        m_bRunning = false;
        if (m_sp4thRecv.get()) {
            pthread_cancel(m_sp4thRecv->native_handle()); // 强制停止线程
            if (m_sp4thRecv->joinable()) {
                m_sp4thRecv->join();
            }
            m_sp4thRecv.reset();
            m_sp4thRecv = nullptr;
        }

        if (m_bReconnecting) {
            m_bReconnecting = false;
            if (m_sp4thReconn.get()) {
                pthread_cancel(m_sp4thReconn->native_handle()); // 强制停止线程
                if (m_sp4thReconn->joinable()) {
                    m_sp4thReconn->join();
                }
                m_sp4thReconn.reset();
                m_sp4thReconn = nullptr;
            }
        }

        if (m_pReadBuffer) {
            delete[] m_pReadBuffer;
            m_pReadBuffer = nullptr;
        }

        m_bReconnecting = false;
        m_bConnected = false;
        m_bInited = false;
    }

    return;
}

bool TcpClient::connect()
{
    bool bRet = false;
    try {

        if (!(m_sp4ObjIOCtx.get()))
        {
            m_sp4ObjIOCtx = std::make_shared<asio::io_context>();
        }

        if (!(m_sp4ObjSocket.get()))
        {
            m_sp4ObjSocket = std::make_shared<asio::ip::tcp::socket>(*(m_sp4ObjIOCtx.get()));
        }

        asio::ip::tcp::resolver resolver(*(m_sp4ObjIOCtx.get()));
        m_objRemoteEndPoint = asio::connect(*(m_sp4ObjSocket.get())
                                            , resolver.resolve(m_stConnCfgInfo.strSrvIP
                                                               , std::to_string(m_stConnCfgInfo.ui16SrvPort)));

        //std::error_code errCode;
        //m_objRemoteEndPoint = asio::ip::tcp::endpoint(asio::ip::address_v4::from_string(m_stConnCfgInfo.strSrvIP), m_stConnCfgInfo.ui16SrvPort);
        //m_sp4ObjSocket->connect(m_objRemoteEndPoint, errCode);

        //auto endpoints = asio::ip::tcp::resolver(*(m_sp4ObjIOCtx.get())).resolve(m_stConnCfgInfo.strSrvIP, std::to_string(m_stConnCfgInfo.ui16SrvPort));
        //asio::async_connect(*(m_sp4ObjIOCtx.get()), endpoints, close_after(std::chrono::seconds(10), *(m_sp4ObjIOCtx.get())));

        m_sp4ObjIOCtx->run();
        m_bConnected = true;

        // 先停止数据read线程，避免deinit退出阻塞
        m_bRunning = false;
        if (m_sp4thRecv.get()) {
            pthread_cancel(m_sp4thRecv->native_handle()); // 强制停止线程
            if (m_sp4thRecv->joinable()) {
                m_sp4thRecv->join();
            }
            m_sp4thRecv.reset();
            m_sp4thRecv = nullptr;
        }

        m_bRunning = true;
        if (!(m_sp4thRecv.get()))
        {
            m_sp4thRecv = std::make_shared<std::thread> (&TcpClient::onRecvDataProcess, this);
        }

        m_bReconnecting = false;
        m_bInited = true;
        bRet = true;

    } catch (std::exception& e) {
        printf("func [TcpClient::connect], connect to server (%s:%d) failed, error info: %s.\r\n" \
               , m_stConnCfgInfo.strSrvIP.data(), m_stConnCfgInfo.ui16SrvPort, e.what());
    }

    return bRet;
}

uint32_t TcpClient::read(char* pDataBuf, uint32_t ui32DataBufLen)
{
    uint32_t ui32Ret = 0;
    try {
        if (m_sp4ObjSocket.get() && pDataBuf)
        {
#if 0
            std::size_t s32Ret = asio::read(*(m_sp4ObjSocket.get())
                                            , asio::buffer(pDataBuf, static_cast<std::size_t>(ui32DataBufLen)));
#else
            std::size_t s32Ret = m_sp4ObjSocket->read_some(asio::buffer(pDataBuf, static_cast<std::size_t>(ui32DataBufLen)));
#endif
            ui32Ret = static_cast<uint32_t>(s32Ret);
        }
    } catch (std::exception& e) {
        std::cerr << "func [TcpClient::read], exception info: " << e.what() << std::endl;
        m_bConnected = false;

        if (m_stConnCfgInfo.bAutoReconn) {
            reconnectToServer();
        }
    }

    return ui32Ret;
}

uint32_t TcpClient::read(std::string& strDataBuf)
{
#if 0
    uint32_t ui32Ret = 0;
    if (m_sp4ObjSocket.get())
    {
        std::size_t s32Ret = asio::read(*(m_sp4ObjSocket.get()), asio::buffer(strDataBuf));
        ui32Ret = static_cast<uint32_t>(s32Ret);
    }
    return ui32Ret;
#else
    char *pBuffer = new (std::nothrow) char[m_stConnCfgInfo.ui32BufMaxSize];
    if (!pBuffer)
    {
        return 0;
    }

    memset(pBuffer, 0, m_stConnCfgInfo.ui32BufMaxSize);
    uint32_t ui32Ret = read(pBuffer, m_stConnCfgInfo.ui32BufMaxSize);
    if (0 < ui32Ret)
    {
        strDataBuf = std::string(pBuffer, ui32Ret);
    }

    if (pBuffer)
    {
        delete[] pBuffer;
        pBuffer = nullptr;
    }
    return ui32Ret;
#endif
}

uint32_t TcpClient::write(const char* pDataBuf, uint32_t ui32DataBufLen)
{
    uint32_t ui32Ret = 0;

    try {
        if (m_sp4ObjSocket.get() && pDataBuf)
        {
#if 0
            if (m_stConnCfgInfo.bAsync)
            {
                asio::async_write(*(m_sp4ObjSocket.get()), asio::buffer(pDataBuf, static_cast<std::size_t>(ui32DataBufLen))
                                  , close_after(std::chrono::seconds(10), *(m_sp4ObjSocket.get())));
            }
#endif
            std::size_t s32Ret = asio::write(*(m_sp4ObjSocket.get())
                                             , asio::buffer(pDataBuf, static_cast<std::size_t>(ui32DataBufLen)));
            ui32Ret = static_cast<uint32_t>(s32Ret);
        }
    } catch (std::exception& e) {
        std::cerr << "func [TcpClient::write], exception info: " << e.what() << std::endl;
        m_bConnected = false;
    }

    return ui32Ret;
}

uint32_t TcpClient::write(const std::string& strDataBuf)
{
    uint32_t ui32Ret = 0;

    try {
        if (m_sp4ObjSocket.get())
        {
#if 0
            std::size_t s32Ret = asio::write(*(m_sp4ObjSocket.get()), asio::buffer(strDataBuf));
#else
            std::error_code errorCode;
            std::size_t s32Ret = m_sp4ObjSocket->write_some(asio::buffer(strDataBuf), errorCode);
#endif
            ui32Ret = static_cast<uint32_t>(s32Ret);
        }
    } catch (std::exception& e) {
        std::cerr << "func [TcpClient::write], exception info: " << e.what() << std::endl;
        m_bConnected = false;
    }

    return ui32Ret;
#if 0
    return write(strDataBuf.data(), static_cast<uint32_t>(strDataBuf.size()));
#endif
}

std::string TcpClient::remoteInfo()
{
    return BaseConnection::remoteInfo(m_objRemoteEndPoint);
}

void TcpClient::asyncRead()
{
    if (!m_bConnected) {
        return;
    }

    if (m_pReadBuffer) {
        memset(m_pReadBuffer, 0, m_stConnCfgInfo.ui32BufMaxSize);

        try {
            //auto self(shared_from_this());
            m_sp4ObjSocket->async_read_some(asio::buffer(m_pReadBuffer
                                                         , static_cast<std::size_t>(m_stConnCfgInfo.ui32BufMaxSize)),
                                            [this](std::error_code ec, std::size_t length)
            {
                (void)this;
                if (!ec) {
                    std::string strDataInfo(m_pReadBuffer, length);
                    if (!(strDataInfo.empty())) {
                        if (m_pfnRecvCallback) {
                            m_pfnRecvCallback(strDataInfo);
                        }
                    }
                } else {
                    std::cerr << "func [asio::ip::tcp::socket::async_read_some], read failed: " << ec.message() << std::endl;
                    m_bConnected = false;

                    if (m_stConnCfgInfo.bAutoReconn) {
                        reconnectToServer();
                    }
                }

                return;
            });
        } catch (std::exception& e) {
            std::cerr << "func [TcpClient::onRecvDataProcess], exception info: " << e.what() << std::endl;
            m_bConnected = false;

            if (m_stConnCfgInfo.bAutoReconn) {
                reconnectToServer();
            }
        }
    }

    return;
}

void TcpClient::onRecvDataProcess()
{
    while (m_bRunning)
    {
        if (m_bConnected)
        {
            // 同步模式
            std::string strDataInfo = "";
            read(strDataInfo);

            if (!(strDataInfo.empty()))
            {
                if (m_pfnRecvCallback)
                {
                    m_pfnRecvCallback(strDataInfo);
                }
            }
        }

        std::this_thread::yield();
    }

    return;
}

void TcpClient::reconnectToServer()
{
    if (m_sp4thReconn.get()) {
        m_bReconnecting = false;

        if (m_sp4thReconn->joinable()) {
            m_sp4thReconn->join();
        }
        m_sp4thReconn.reset();
        m_sp4thReconn = nullptr;
    }

    m_bReconnecting = true;
    m_sp4thReconn = std::make_shared<std::thread> ([this]() {
        while (m_bReconnecting) {
            std::cout << "func [TcpClient::reconnectToServer], reconnet to server." << std::endl;
            if (connect()) {
                break;
            }
            m_bInited = true; // 使能deinit正确执行
            std::this_thread::sleep_for(std::chrono::milliseconds(2 * 1000));
        }
        std::cout << "func [TcpClient::reconnectToServer], exit reconnet to server thread." << std::endl;
        return;
    });

    if (m_sp4thReconn.get()) {
        pthread_setname_np(m_sp4thReconn->native_handle(), "TcpClient::reconnectToServer");
    }

    return;
}


} // namespace NetworkUtil

